package com.nucarf.bi.digtalize.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.nucarf.bi.digtalize.bean.*;
import com.nucarf.bi.digtalize.util.FeiShuWarnUtil;
import com.nucarf.bi.digtalize.util.PropertyUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.Date;

@Slf4j
public class FlatMap2BinLogFunction extends RichFlatMapFunction<JSONObject, BinLogRecord> {

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void flatMap(JSONObject value, Collector<BinLogRecord> out) throws Exception {
        try{
            BinLogRecord logRecord = null;
            String table = value.getString("table");
            String before = value.getString("before");
            String after = value.getString("after");
            String op_type = value.getString("op_type");
            String database = value.getString("database");
            if (op_type.equals("insert")) {
                switch (table) {
                    case "global_company_order":
                        logRecord = new BinLogRecord<GlobalConsume>();
                        GlobalConsume beforeConsume = JSON.parseObject(before, GlobalConsume.class);
                        GlobalConsume afterConsume = JSON.parseObject(after, GlobalConsume.class);
                        logRecord.setBefore(beforeConsume);
                        logRecord.setAfter(afterConsume);
                        break;
                    case "global_company_recharge":
                        logRecord = new BinLogRecord<GlobalRecharge>();
                        GlobalRecharge beforeRecharge = JSON.parseObject(before, GlobalRecharge.class);
                        GlobalRecharge afterRecharge = JSON.parseObject(after, GlobalRecharge.class);
                        logRecord.setBefore(beforeRecharge);
                        logRecord.setAfter(afterRecharge);
                        break;
                    case "global_company_balance":
                        logRecord = new BinLogRecord<GlobalBalance>();
                        GlobalBalance beforeBalance = JSON.parseObject(before, GlobalBalance.class);
                        GlobalBalance afterBalance = JSON.parseObject(after, GlobalBalance.class);
                        logRecord.setBefore(beforeBalance);
                        logRecord.setAfter(afterBalance);
                        break;
                }
                logRecord.setDatabase(database);
                logRecord.setTable(table);
                logRecord.setOpType(op_type);
                logRecord.setPrimkey(value.getString("primkey"));
                logRecord.setTs(value.getString("ts"));
                out.collect(logRecord);
            } else {
                log.warn("error op_type database={},table={}, json={}", database, table, value.toJSONString());
                String warnMsg = "bin-log 异常操作 database=" + database + ",table=" + table + ",opType=" + op_type + ",json=" + value.toJSONString();
                WarnMessage warnMessage = new WarnMessage(PropertyUtil.getPropertyValue("feishu.warn.evn"), warnMsg, new Date().toString());
                FeiShuWarnUtil.sendFeishuWarning(JSON.toJSONString(warnMessage));
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
